package com.icreate.storm.methods.mongo;

import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.bson.Document;

import java.util.HashSet;
import java.util.Set;

/**
 * Created by hadoop on 17-5-2.
 */
public class HourINCMongoMapper implements MongoMapper {
    private Set<String> excludeFields;

    public HourINCMongoMapper(String... excludeFields) {
        Set<String> idFields = new HashSet<>();
        for (String excludeField : excludeFields) {
            idFields.add(excludeField);
        }
        this.excludeFields = idFields;
    }

    public Document toDocument(ITuple tuple) {
        String hour = tuple.getValueByField("hour").toString();

        Document document = new Document();
        Document incDocument = new Document();
        Fields fields = tuple.getFields();
        for (String field : fields) {
            if (!excludeFields.contains(field))
                incDocument.append("hour." + hour + "." + field, tuple.getValueByField(field));
        }
        document.append("$inc", incDocument);
        return document;
    }
}

